package com.flink.datastreamapi.transformation;


import com.flink.datastreamapi.bean.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TransFilter {

    public static void main(String[] args) throws Exception {

        //获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //读取数据源
        DataStreamSource<WaterSensor> dataStreamSource = env.fromElements(
                new WaterSensor("sensor_1", 1l, 1),
                new WaterSensor("sensor_2", 2l, 2),
                new WaterSensor("sensor_3", 3l, 3),
                new WaterSensor("sensor_1", 1l, 1)
                );

        //数据转换--过滤数据,过滤掉不符合条件的数据
        //方式一： 匿名内部类
        dataStreamSource.filter(new FilterFunction<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor value) throws Exception {
                //若为`true`则元素正常输出，若为`false`则元素被过滤掉
                //过滤掉id不为sensor_1的WaterSensor对象，

                return  value.id.equals("sensor_1");
            }
        })
        //输出
        .print();

        //方式二: 传入FilterFunction实现类
        dataStreamSource.filter(new CustomFilter()).print();



        //执行
        env.execute();
    }

    public static class CustomFilter implements FilterFunction<WaterSensor>{
        @Override
        public boolean filter(WaterSensor value) throws Exception {
            //若为`true`则元素正常输出，若为`false`则元素被过滤掉
            //过滤掉id不为sensor_1的WaterSensor对象《

            return  value.id.equals("sensor_1");
        }
    }
}
